package com.sixiaojie.flink.scheduler;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.javafaker.Faker;
import com.sixiaojie.flink.domain.*;
import com.sixiaojie.flink.dto.*;
import com.sixiaojie.flink.repository.*;
import lombok.Data;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 数据生成调度器
 * 实现用户行为->订单->支付->物流的完整业务流程
 */
@Component
public class DataGeneratorScheduler implements CommandLineRunner {

    // 仓库接口
    @Resource
    private UserRepository userRepository;
    @Resource
    private ProductRepository productRepository;
    @Resource
    private UserBehaviorSummaryRepository userBehaviorSummaryRepository;
    @Resource
    private PaymentRepository paymentRepository;
    @Resource
    private OrderRepository orderRepository;
    @Resource
    private LogisticsRepository logisticsRepository;

    // Kafka模板
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 工具类
    private final Faker faker = new Faker(Locale.CHINA);
    private final Random random = new Random();
    private final ObjectMapper objectMapper = new ObjectMapper();

    // 静态配置
    private static final List<String> CATEGORIES = Arrays.asList(
            "手机数码", "电脑办公", "家用电器", "食品生鲜",
            "美妆护肤", "服饰鞋包", "家居家装", "母婴玩具"
    );
    private static final List<String> DEVICE_TYPES = Arrays.asList(
            "PC", "iOS", "Android", "H5"
    );
    private static final List<String> PAYMENT_METHODS = Arrays.asList(
            "支付宝", "微信支付", "信用卡", "银联支付"
    );

    // 运行时状态
    private List<Long> userIds;
    private List<Product> products;
    private final Map<Long, String> userSessions = new ConcurrentHashMap<>();
    private final Map<String, OrderProcessContext> orderProcessMap = new ConcurrentHashMap<>();

    /**
     * 应用启动时初始化检查
     */
    @Override
    @Transactional
    public void run(String... args) {
        checkAndInitializeData();
    }
    /**
     * 检查并初始化基础数据
     */
    private void checkAndInitializeData() {
        // 检查用户数据
        long userCount = userRepository.count();
        if (userCount == 0) {
            System.out.println("检测到用户数据为空，正在初始化基础数据...");
            generateBaseData();
            System.out.println("基础数据初始化完成！");
        } else {
            initializeRuntimeData();
        }
    }

    /**
     * 每天凌晨生成基础数据
     */
    @Scheduled(cron = "0 0 0 * * ?")
    @Transactional
    public void generateBaseData() {
        // 生成5个新用户
        generateUsers(5);
        // 生成3个新产品
        generateProducts(3);
        // 初始化运行时数据
        initializeRuntimeData();
    }

    /**
     * 每分钟生成/更新用户行为汇总数据（按用户按天累加）
     */
    @Scheduled(fixedRate = 60000)
    @Transactional
    public void generateUserBehaviorSummary() {
        LocalDate yesterday = LocalDate.now().minusDays(1);
        String dateStr = yesterday.toString();

        userRepository.findAll().forEach(user -> {
            // 查询或创建当天的汇总记录
            UserBehaviorSummary summary = userBehaviorSummaryRepository
                    .findByUserIdAndDate(user.getId(), dateStr)
                    .orElseGet(() -> {
                        UserBehaviorSummary newSummary = new UserBehaviorSummary();
                        newSummary.setUserId(user.getId());
                        newSummary.setDate(dateStr);
                        return newSummary;
                    });

            // 累加行为数据（模拟增量）
            summary.setViewCount(summary.getViewCount() + random.nextInt(10));
            summary.setClickCount(summary.getClickCount() + random.nextInt(6));
            summary.setSearchCount(summary.getSearchCount() + random.nextInt(4));
            summary.setCartAddCount(summary.getCartAddCount() + random.nextInt(2));
            summary.setPurchaseCount(summary.getPurchaseCount() +
                    (random.nextDouble() < 0.3 ? 1 : 0)); // 30%概率增加购买次数

            userBehaviorSummaryRepository.save(summary);
        });
    }

    /**
     * 每10秒生成实时业务数据
     */
    @Scheduled(fixedRate = 10000)
    @Transactional
    public void generateRealTimeData() {

        // 1. 生成用户行为（可能触发下单）
        generateUserBehaviors(5 + random.nextInt(11));

        // 2. 处理待支付订单
        processPendingOrders();

        // 3. 处理已支付订单（发货）
        processPaidOrders();

        // 4. 更新物流状态
        updateLogisticsStatus();
    }

    /**
     * 生成用户行为数据
     */
    private void generateUserBehaviors(int count) {
        if (userSessions.isEmpty()) return;

        List<Long> activeUserIds = new ArrayList<>(userSessions.keySet());

        for (int i = 0; i < count; i++) {
            Long userId = activeUserIds.get(random.nextInt(activeUserIds.size()));
            UserBehaviorEvent event = createBehaviorEvent(userId);

            // 5%概率触发下单
            if ("PURCHASE".equals(event.getBehaviorType())) {
                createOrder(userId);
            }

            // 发送行为事件
            kafkaTemplate.send("user-behaviors",
                    event.getUserId().toString(),
                    event
            );
        }
    }

    /**
     * 创建完整订单流程
     */
    private void createOrder(Long userId) {
        // 1. 选择商品和数量
        Product product = products.get(random.nextInt(products.size()));
        int quantity = 1 + random.nextInt(5);
        BigDecimal amount = product.getPrice().multiply(BigDecimal.valueOf(quantity));

        // 2. 创建订单
        Orders order = new Orders();
        order.setId(UUID.randomUUID().toString());
        order.setUserId(userId);
        order.setProductId(product.getId());
        order.setQuantity(quantity);
        order.setAmount(amount);
        order.setStatus("CREATED");
        order.setCreatedAt(new Date());
        orderRepository.save(order);

        // 3. 扣减库存
        updateProductStock(product.getId(), -quantity);

        // 4. 创建支付记录
        Payment payment = createPayment(order);

        // 5. 初始化物流
        Logistics logistics = createLogistics(order);

        // 6. 保存流程上下文
        orderProcessMap.put(order.getId(),
                new OrderProcessContext(order, payment, logistics)
        );

        // 7. 发送订单创建事件
        OrderEvent orderEvent = new OrderEvent();
        orderEvent.setOrderId(order.getId());
        orderEvent.setUserId(order.getUserId());
        orderEvent.setProductId(order.getProductId());
        orderEvent.setQuantity(order.getQuantity());
        orderEvent.setAmount(order.getAmount());
        orderEvent.setStatus(order.getStatus());

        kafkaTemplate.send("order-events", order.getId(), orderEvent);
    }

    /**
     * 处理待支付订单
     */
    private void processPendingOrders() {
        orderProcessMap.values().stream()
                .filter(ctx -> "CREATED".equals(ctx.getOrder().getStatus()))
                .forEach(ctx -> {
                    // 80%概率支付成功
                    if (random.nextDouble() < 0.8) {
                        // 更新支付状态
                        ctx.getPayment().setStatus("SUCCESS");
                        ctx.getPayment().setTransactionTime(new Date());
                        paymentRepository.save(ctx.getPayment());

                        // 更新订单状态
                        ctx.getOrder().setStatus("PAID");
                        orderRepository.save(ctx.getOrder());

                        // 发送支付成功事件
                        PaymentEvent paymentEvent = new PaymentEvent();
                        paymentEvent.setPaymentId(ctx.getPayment().getId());
                        paymentEvent.setOrderId(ctx.getOrder().getId());
                        paymentEvent.setStatus(ctx.getPayment().getStatus());
                        paymentEvent.setAmount(ctx.getPayment().getAmount());

                        kafkaTemplate.send("payment-events",
                                ctx.getOrder().getId(),
                                paymentEvent
                        );
                    } else {
                        // 支付失败
                        ctx.getPayment().setStatus("FAILED");
                        paymentRepository.save(ctx.getPayment());

                        // 从流程中移除
                        orderProcessMap.remove(ctx.getOrder().getId());
                    }
                });
    }

    /**
     * 处理已支付订单（发货）
     */
    private void processPaidOrders() {
        orderProcessMap.values().stream()
                .filter(ctx -> "PAID".equals(ctx.getOrder().getStatus()))
                .forEach(ctx -> {
                    // 更新订单状态
                    ctx.getOrder().setStatus("SHIPPED");
                    orderRepository.save(ctx.getOrder());

                    // 更新物流状态
                    ctx.getLogistics().setStatus("SHIPPED");
                    ctx.getLogistics().setLocation("已发货");
                    ctx.getLogistics().setShippingTime(new Date());
                    logisticsRepository.save(ctx.getLogistics());

                    // 发送物流事件
                    LogisticsEvent logisticsEvent = new LogisticsEvent();
                    logisticsEvent.setLogisticsId(ctx.getLogistics().getId());
                    logisticsEvent.setOrderId(ctx.getOrder().getId());
                    logisticsEvent.setStatus(ctx.getLogistics().getStatus());
                    logisticsEvent.setLocation(ctx.getLogistics().getLocation());

                    kafkaTemplate.send("logistics-events",
                            ctx.getOrder().getId(),
                            logisticsEvent
                    );
                });
    }

    /**
     * 更新物流状态
     */
    private void updateLogisticsStatus() {
        orderProcessMap.values().stream()
                .filter(ctx -> "SHIPPED".equals(ctx.getLogistics().getStatus()))
                .forEach(ctx -> {
                    String[] statusFlow = {"IN_TRANSIT", "IN_TRANSIT", "DELIVERED"};
                    String nextStatus = getNextStatus(ctx.getLogistics().getStatus(), statusFlow);

                    if (nextStatus != null) {
                        // 更新物流状态
                        ctx.getLogistics().setStatus(nextStatus);
                        ctx.getLogistics().setLocation(faker.address().cityName());

                        if ("DELIVERED".equals(nextStatus)) {
                            ctx.getLogistics().setActualDelivery(new Date());
                            ctx.getOrder().setStatus("COMPLETED");
                            orderRepository.save(ctx.getOrder());

                            // 流程结束，移除上下文
                            orderProcessMap.remove(ctx.getOrder().getId());
                        }

                        logisticsRepository.save(ctx.getLogistics());

                        // 发送物流更新事件
                        LogisticsEvent logisticsEvent = new LogisticsEvent();
                        logisticsEvent.setLogisticsId(ctx.getLogistics().getId());
                        logisticsEvent.setOrderId(ctx.getOrder().getId());
                        logisticsEvent.setStatus(ctx.getLogistics().getStatus());
                        logisticsEvent.setLocation(ctx.getLogistics().getLocation());

                        kafkaTemplate.send("logistics-events",
                                ctx.getOrder().getId(),
                                logisticsEvent
                        );
                    }
                });
    }

    // ========== 工具方法 ==========

    private void generateUsers(int count) {
        List<User> newUsers = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            User user = new User();
            user.setName(faker.name().fullName());
            user.setAge(faker.number().numberBetween(18, 70));
            user.setGender(faker.options().option(0, 1));
            user.setRegion(faker.address().cityName());
            newUsers.add(user);
        }
        userRepository.saveAll(newUsers);
        System.out.println("已生成 " + count + " 个新用户");
    }

    private void generateProducts(int count) {
        List<Product> newProducts = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            Product product = new Product();
            product.setName(generateChineseProductName());
            product.setCategory(CATEGORIES.get(random.nextInt(CATEGORIES.size())));
            product.setStock(faker.number().numberBetween(100, 1000));
            product.setPrice(BigDecimal.valueOf(
                    faker.number().randomDouble(2, 10, 10000)
            ));
            newProducts.add(product);
        }
        productRepository.saveAll(newProducts);
        System.out.println("已生成 " + count + " 个新商品");
    }

    private String generateChineseProductName() {
        String[] brands = {"华为", "小米", "苹果", "联想", "海尔"};
        String[] types = {"智能手机", "笔记本电脑", "平板电脑", "智能手表"};
        String[] specs = {"Pro", "Max", "旗舰版", "尊享版"};

        return String.format("%s%s %s",
                brands[random.nextInt(brands.length)],
                specs[random.nextInt(specs.length)],
                types[random.nextInt(types.length)]
        );
    }

    private void initializeRuntimeData() {
        this.userIds = userRepository.findAll()
                .stream()
                .map(User::getId)
                .collect(Collectors.toList());

        this.products = productRepository.findAll();

        userIds.forEach(userId ->
                userSessions.put(userId, UUID.randomUUID().toString())
        );
        System.out.println("已初始化运行时数据：用户数=" + userIds.size() +
                ", 商品数=" + products.size());
    }

    private UserBehaviorEvent createBehaviorEvent(Long userId) {
        UserBehaviorEvent event = new UserBehaviorEvent();
        event.setUserId(userId);
        event.setBehaviorType(getRandomBehaviorType());
        event.setTargetId(generateTargetId(event.getBehaviorType()));
        event.setDeviceType(DEVICE_TYPES.get(random.nextInt(DEVICE_TYPES.size())));
        event.setSessionId(userSessions.get(userId));
        event.setIpAddress(faker.internet().ipV4Address());

        if (random.nextDouble() < 0.3) {
            event.setExtraData(generateExtraData(event.getBehaviorType()));
        }

        return event;
    }

    private Payment createPayment(Orders order) {
        Payment payment = new Payment();
        payment.setId(UUID.randomUUID().toString());
        payment.setOrderId(order.getId());
        payment.setUserId(order.getUserId());
        payment.setAmount(order.getAmount());
        payment.setPaymentMethod(PAYMENT_METHODS.get(
                random.nextInt(PAYMENT_METHODS.size())
        ));
        payment.setStatus("PENDING");
        paymentRepository.save(payment);
        return payment;
    }

    private Logistics createLogistics(Orders order) {
        Logistics logistics = new Logistics();
        logistics.setId(UUID.randomUUID().toString());
        logistics.setOrderId(order.getId());
        logistics.setStatus("PROCESSING");
        logistics.setLocation("仓库");
        logistics.setEstimatedDelivery(
                faker.date().future(3, TimeUnit.DAYS)
        );
        logisticsRepository.save(logistics);
        return logistics;
    }

    private String getRandomBehaviorType() {
        double r = random.nextDouble();
        if (r < 0.5) return "VIEW";
        if (r < 0.7) return "CLICK";
        if (r < 0.85) return "SEARCH";
        if (r < 0.95) return "ADD_TO_CART";
        return "PURCHASE";
    }

    private String generateTargetId(String behaviorType) {
        switch (behaviorType) {
            case "VIEW":
            case "CLICK":
            case "ADD_TO_CART":
            case "PURCHASE":
                return "PROD_" + (10000 + random.nextInt(90000));
            case "SEARCH":
                return "KEYWORD_" + faker.lorem().word();
            default:
                return "PAGE_" + faker.internet().slug();
        }
    }

    private String generateExtraData(String behaviorType) {
        try {
            Map<String, Object> extra = new HashMap<>();
            switch (behaviorType) {
                case "CLICK":
                    extra.put("positionX", random.nextInt(1000));
                    extra.put("positionY", random.nextInt(800));
                    break;
                case "SEARCH":
                    extra.put("keyword", faker.lorem().word());
                    extra.put("resultsCount", random.nextInt(100));
                    break;
                case "ADD_TO_CART":
                    extra.put("quantity", 1 + random.nextInt(3));
                    break;
            }
            return extra.isEmpty() ? null : objectMapper.writeValueAsString(extra);
        } catch (Exception e) {
            return null;
        }
    }

    private String getNextStatus(String currentStatus, String[] flow) {
        for (int i = 0; i < flow.length - 1; i++) {
            if (flow[i].equals(currentStatus)) {
                return flow[i + 1];
            }
        }
        return null;
    }

    private void updateProductStock(Long productId, int delta) {
        productRepository.findById(productId).ifPresent(product -> {
            product.setStock(product.getStock() + delta);
            productRepository.save(product);
        });
    }


    /**
     * 订单流程上下文
     */
    @Data
    private static class OrderProcessContext {
        private final Orders order;
        private final Payment payment;
        private final Logistics logistics;

        public OrderProcessContext(Orders order, Payment payment, Logistics logistics) {
            this.order = order;
            this.payment = payment;
            this.logistics = logistics;
        }
    }
}